package manager

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"os"
	"path/filepath"
	"sieve_engine/execute"
	"sieve_engine/task"
	"sieve_engine/utils"
	"strconv"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	"github.com/go-redis/redis/v8"
	"github.com/spf13/cast"
	"sieve_engine/lib/config"
	"sieve_engine/lib/log"
	"sieve_engine/lib/net/ip"
)

var (
	timeFormat            = "2006-01-02 15:04:05"
	timeFormatMillisecond = "2006-01-02 15:04:05.999999999"

	s           = *config.String("sieve.redis.channel", "material_task_queue,material_task_detail", "redis channel")
	ss          = strings.Split(s, ",")
	redisParams = RedisParams{
		Addr:          *config.String("sieve.redis.host", "", "redis host"),
		Pwd:           *config.String("sieve.redis.pwd", "", "redis pwd"),
		DB:            *config.Int("sieve.redis.db", 0, "redis db"),
		Prefix:        *config.String("sieve.redis.prefix", "", "redis prefix"),
		Separator:     *config.String("sieve.redis.separator", "", "redis separator"),
		Channel:       ss[0],
		DetailChannel: ss[1],
	}
	clusterID       = *config.String("sieve.cluster.id", "", "sieve cluster id")
	initOnce        = sync.Once{}
	sieveResultPath = *config.String("sieve.file_path", "/data/sieve_engine", "sieve file path")
)

const (
	uploadTypeFormFile = "file"
	uploadTypeFormArg  = "arg"
)

const (
	fileTypeSrc    = "src"
	fileTypeResult = "result"
)

type TaskDistributor struct {
	BatchSize  int // 一共多少个
	ChunkSize  int // 任务分片执行的每片多少个
	ChunkCount int // 任务分片执行的有多少片（对应多少个协程）
}

func getDumpFilePath(taskId, ty string) string {

	msec, _ := strconv.ParseUint(taskId, 10, 64)
	t := time.UnixMilli(int64(msec))
	tt := fmt.Sprintf("%v%v%v", t.Year(), int(t.Month()), t.Day())
	fileAbsPath := filepath.Join(sieveResultPath, tt)
	if !utils.Exists(fileAbsPath) {
		_ = os.Mkdir(fileAbsPath, 0644)
	}

	return filepath.Join(fileAbsPath, fmt.Sprintf("%v_%v.txt", taskId, ty))
}

type TaskManagerService struct {
	clusterServer *ClusterServer
	taskConfig    *TaskConfig
	locker        sync.Mutex
	taskOp        *atomic.Bool
	redisParams   RedisParams
	redisCli      *redis.Client
	executors     map[string]*TaskExecutorService
	monitors      map[string]*Counter
}

func NewTaskManagerService() *TaskManagerService {

	ts := &TaskManagerService{
		taskConfig:  getTaskConfig(),
		executors:   make(map[string]*TaskExecutorService),
		redisParams: redisParams,
		taskOp:      &atomic.Bool{},
	}

	initOnce.Do(func() {
		ts.lazyInit()
		log.Info(fmt.Sprintf("已经完成 REDIS 连接初始化工作"))
	})

	return ts
}

func (manager *TaskManagerService) lazyInit() {

	manager.redisCli = redis.NewClient(&redis.Options{
		Addr:         manager.redisParams.Addr,
		Password:     manager.redisParams.Pwd, // 如果有密码，请填写对应的密码
		DB:           manager.redisParams.DB,  // 选择要使用的数据库
		WriteTimeout: time.Minute * 5,
		ReadTimeout:  time.Minute * 5,
		DialTimeout:  time.Minute * 5,
	})
	manager.initClusterServer()
}

func (manager *TaskManagerService) initClusterServer() {

	var (
		members []string
		port    = utils.RandInt(30000, 60000)
		address = fmt.Sprintf("%v:%v", ip.GetLocalIP(), port)
	)

	for _, v := range getClusterConfig().Node {
		if v.ID == clusterID {
			address = v.Address
			continue
		}
		members = append(members, v.Address)
	}

	var (
		option = &Options{
			Address: address,
			ID:      clusterID,
			Members: members,
		}
		clusterNode = ClusterConfig{
			Node: []ClusterConfigItem{
				{
					ID:      option.ID,
					Address: option.Address,
				},
			},
		}
	)

	cs, err := NewClusterServer(option)
	if err != nil {
		panic(fmt.Sprintf("创建集群节点失败:%v", err))
	}

	manager.clusterServer = cs

	/* 1. 更新节点信息 */
	_ = setClusterConfig(clusterNode)

	/* 2. 设置集群的通知事件回调 */
	manager.setClusterNotifyCallBack()
}

func (manager *TaskManagerService) setClusterNotifyCallBack() {

	manager.clusterServer.SetNotifyCallBack(manager.clusterEventNotify)
}

func (manager *TaskManagerService) clusterEventNotify(action, key string, val interface{}) {

	if action == "set" {
		switch key {
		case "running_tasks":
			v := cast.ToInt(val)
			manager.taskConfig.runningTasks.Add(int32(v))
			break
		case "concurrency":
			v := cast.ToInt(val)
			manager.taskConfig.concurrency = int32(v)
			break
		}
	}
}

func (manager *TaskManagerService) Create(req *CreateReq, fCallBack func(f string) error) (*CreateRsp, error) {

	var (
		value       []string
		srcFilePath string
	)

	req.TaskID = fmt.Sprintf("%v", time.Now().UnixMilli())
	if req.UploadType == uploadTypeFormFile {
		if fCallBack != nil {
			srcFilePath = getDumpFilePath(req.TaskID, fileTypeSrc)
			_ = fCallBack(srcFilePath)
		}
	}

	if req.UploadType == uploadTypeFormArg {

		for _, line := range strings.Split(req.Phones, ",") {
			/*
				1. 过滤前后空格
				2. 过滤非手机号
				3. 暂时不支持带 "-" 的料子，如果后面要支持，不能用 strconv.ParseUint 过滤
			*/

			lineStr := strings.TrimSpace(line)
			value = append(value, lineStr)
		}
	}

	go func() {

		if req.UploadType == uploadTypeFormFile {
			if bitmap, _, err := utils.NewBitMapFromFile(srcFilePath, false); err != nil {
				log.Err(fmt.Sprintf("读取筛料文件失败:%+v", err))
			} else {
				for _, v := range bitmap.ToArray() {
					value = append(value, fmt.Sprintf("%v", v))
				}
			}
		}

		start := time.Now()
		if _, err := manager.create(value, req); err != nil {
			log.Err(fmt.Sprintf("创建任务失败：%+v", err))
		} else {
			if req.UploadType == uploadTypeFormFile {
				_ = os.Remove(srcFilePath)
			}
		}
		log.Info(fmt.Sprintf("任务:%v 上传文件耗时:%v", req.TaskID, time.Since(start).String()))
	}()

	taskCount := len(value)
	return &CreateRsp{
		TaskID:       req.TaskID,
		TotalCount:   taskCount,
		ValidCunt:    taskCount,
		InValidCount: taskCount,
	}, nil
}

func (manager *TaskManagerService) create(value []string, req *CreateReq) (*CreateRsp, error) {

	var (
		chunkSize = 200
		ctx       = context.Background()
		values    = utils.ChunkSliceString(value, chunkSize)
		taskCount = len(value)
	)

	// 1. 添加redis任务
	id := req.TaskID
	now := time.Now()
	taskInfo := task.Info{
		Id:            id,
		Operator:      req.Operator,
		Country:       req.Country,
		AreaCode:      req.AreaCode,
		TaskCount:     taskCount,
		SrcListKey:    fmt.Sprintf("src_list_%v", id),
		ResultListKey: fmt.Sprintf("result_list_%v", id),
		CreateAt:      now,
		UpdateAt:      now,
		Progress: task.Progress{
			State:     task.StatusInit,
			TaskCount: int64(taskCount),
		},
	}

	tx := manager.redisCli.TxPipeline()
	for _, vv := range values {
		tx.SAdd(ctx, taskInfo.SrcListKey, vv)
	}

	b, _ := json.Marshal(&taskInfo)

	/* 任务列表 LIST 头插 */
	if err := tx.LPush(ctx, manager.redisParams.Channel, id).Err(); err != nil {
		log.Err(fmt.Sprintf("任务ID:提交 REDIS 任务失败:%+v", err))
	}

	/* 任务详情 HASH */
	if err := tx.HSet(ctx, manager.redisParams.DetailChannel, id, b).Err(); err != nil {
		log.Err(fmt.Sprintf("任务ID:提交 REDIS 任务失败:%+v", err))
	}

	var (
		retryIndex int
		retryCount = 10
	)

retry1:
	// 提交事务并获取结果
	results, err := tx.Exec(context.Background())
	if err != nil {
		if strings.Contains(err.Error(), "timeout") && retryIndex < retryCount {
			sleepDur := time.Millisecond * time.Duration(10*retryIndex)
			time.Sleep(sleepDur)
			retryIndex = retryIndex + 1
			log.Err(fmt.Sprintf("任务ID:%v 提交Redis事务超时, 休眠%v 继续重试，第%v次重试,总共重试次数:%+v", taskInfo.Id, sleepDur.String(), retryIndex, retryCount))
			goto retry1
		}
		_ = tx.Discard()
		log.Err(fmt.Sprintf("提交Redis事务失败:%+v,", err))
		return nil, err
	}

	// 检查事务执行结果
	for _, result := range results {
		err := result.Err()
		if err != nil {
			log.Err(fmt.Sprintf("提交Redis事务失败:%+v,", err))
			// 回滚事务
			_ = tx.Discard()
			return nil, err
		}
	}

	return &CreateRsp{
		TaskID:       req.TaskID,
		TotalCount:   taskCount,
		ValidCunt:    taskCount,
		InValidCount: taskCount,
	}, nil
}

func (manager *TaskManagerService) createServiceExecutor(taskInfo task.Info) *TaskExecutorService {

	var (
		taskService *TaskExecutorService
	)

	manager.locker.Lock()
	defer manager.locker.Unlock()

retry:
	if _t, ok := manager.executors[taskInfo.Id]; !ok {
		manager.executors[taskInfo.Id] = newTaskExecutorService(taskInfo, manager)
		goto retry
	} else {
		taskService = _t
	}

	return taskService
}

func (manager *TaskManagerService) RunLoop() {

	manager.recovery()
	manager.readTask()
	return
}

func (manager *TaskManagerService) Start(taskId string, normal bool) error {

	/*
		1. 如果是正常启动(接口控制或者创建任务后的默认启动)的情况，manager.taskConfig.runningTasks 变量不可多次 +1
		2. 如果是正常启动，需要判断当前状态是否可启动，是否超过了 manager.taskConfig.concurrency 可同时运行数量的限制
	*/

	if manager.checkWaitingState() {
		return errors.New(ReachedMaxTasks)
	}

	if err := manager.setState(taskId, task.StatusRun, func(service *TaskExecutorService) {

		if (service.taskInfo.State != task.StatusRun && normal) || !normal {
			_ = manager.clusterServer.Set("running_tasks", 1)
			manager.taskConfig.runningTasks.Add(1)
		}
		service.start()
	}); err != nil {
		return err
	}

	return nil
}

func (manager *TaskManagerService) Pause(taskId string) error {

	if err := manager.setState(taskId, task.StatusPause, func(service *TaskExecutorService) {
		if service.taskInfo.State == task.StatusRun {
			_ = manager.clusterServer.Set("running_tasks", -1)
			manager.taskConfig.runningTasks.Add(-1)
		}
		service.pause()
	}); err != nil {
		return err
	}

	return nil
}

// Terminate /* 被外界强行停止的叫 Terminate，自行结束的叫 Finish */
func (manager *TaskManagerService) Terminate(taskId string) error {

	if err := manager.setState(taskId, task.StatusTerminate, func(service *TaskExecutorService) {
		if service.taskInfo.State == task.StatusRun {
			_ = manager.clusterServer.Set("running_tasks", -1)
			manager.taskConfig.runningTasks.Add(-1)
		}
		service.terminate()
	}); err != nil {
		return err
	}

	return nil
}

func (manager *TaskManagerService) finish(taskId string) error {

	var (
		taskInfo task.Info
	)
	/* 任务执行完了会到这里  */
	if err := manager.setState(taskId, task.StatusFinish, func(service *TaskExecutorService) {
		taskInfo = service.taskInfo
		service.finish()
	}); err != nil {
		return err
	} else {
		_ = manager.clusterServer.Set("running_tasks", -1)
		manager.taskConfig.runningTasks.Add(-1)
	}

	return manager.clearRedisTask(taskInfo)
}

func (manager *TaskManagerService) clearRedisTask(taskInfo task.Info) error {

	var (
		now           = time.Now()
		clearAfterSec = 3600
		extraSec      = taskInfo.UpdateAt.Unix() + int64(clearAfterSec) - now.Unix()
		d             = time.Duration(0)
	)

	if extraSec > 0 {
		d = time.Second * time.Duration(extraSec)
	}

	/* 三十分钟后清理任务 */
	time.AfterFunc(d, func() {

		if err := manager.del(taskInfo.Id); err != nil {
			log.Err(fmt.Sprintf("删除任务失败:%+v", taskInfo.Id))
		}
	})
	return nil
}

func (manager *TaskManagerService) Download(taskId string) (string, error) {

	p := getDumpFilePath(taskId, fileTypeResult)
	if !utils.Exists(p) {
		return "", errors.New("文件不存在")
	}

	return p, nil
}

func (manager *TaskManagerService) getTaskExecutorService(taskId string) (*TaskExecutorService, error) {

	manager.locker.Lock()
	defer manager.locker.Unlock()
	_service, ok := manager.executors[taskId]
	if !ok {
		return nil, errors.New(fmt.Sprintf("不存在的任务ID"))
	}

	return _service, nil
}

func (manager *TaskManagerService) Stats(taskId string) (*task.Info, error) {

	return manager.getState(taskId)
}

func (manager *TaskManagerService) checkWaitingState() bool {

	/* 判断是否在控制中 */
	if manager.taskOp.Load() == true {
		return true
	}

	if manager.taskConfig.runningTasks.Load() >= manager.taskConfig.concurrency {
		return true
	}

	return false
}

func (manager *TaskManagerService) recovery() {

	runningTasks, err := manager.getRunningTask()
	if err != nil {
		log.Err(fmt.Sprintf("获取运行中的任务失败:%+v", err))
		return
	}

	var (
		recoveryStateMap = map[task.Status]struct{}{
			task.StatusRun:   {},
			task.StatusInit:  {},
			task.StatusPause: {},
		}
		recoveryRunStateMap = map[task.Status]struct{}{
			task.StatusRun:  {},
			task.StatusInit: {},
		}
		recoveryFinishStateMap = map[task.Status]struct{}{
			task.StatusFinish: {},
		}
		forEach = func() string {
			var ts []string
			for k, _ := range recoveryStateMap {
				ts = append(ts, string(k))
			}
			return strings.Join(ts, ",")
		}
	)
	log.Info(fmt.Sprintf("状态(%v)的任务会设置为启动状态", forEach()))
	for _, taskInfo := range runningTasks {

		executor, ok := execute.Get(taskInfo.Operator)
		if !ok {
			log.Err(fmt.Sprintf("未注册或者没有实现的业务类型 %+v", taskInfo.Operator))
			/* 需要将错误的列表写入到错误列表 */
			continue
		}

		/*
			此处可能需要检查一下任务状态是否正常
			1. taskInfo 的任务已完成数量是否和 结果列表对应
			2. taskInfo 的任务未完成数量是否和 数据源列表对应
		*/

		if _, ok := recoveryStateMap[taskInfo.State]; !ok {
			continue
		}

		/*
			1. Pause 状态只加载到内存中
			2. Init,Run 状态会初始运行
			3. Finish,Terminal 不用加载
		*/
		taskService := manager.createServiceExecutor(taskInfo)

		if _, ok := recoveryFinishStateMap[taskInfo.State]; ok {
			go func() {
				_ = manager.clearRedisTask(taskInfo)
			}()
		}

		if _, ok := recoveryRunStateMap[taskInfo.State]; ok {
			if err := manager.Start(taskInfo.Id, false); err != nil {
				log.Err(fmt.Sprintf("任务:%v 设置任务状态为 启动 失败:%+v", taskInfo.Id, err))
				continue
			}

			log.Info(fmt.Sprintf("恢复的任务:%v %v", taskInfo.Id, taskInfo.State))
			_ = manager.run(executor, taskService)
		}

	}

	return
}

func (manager *TaskManagerService) getRunningTask() ([]task.Info, error) {

	var (
		pNo          = 0
		pSize        = 3
		channel      = fmt.Sprintf("%v_running", manager.redisParams.Channel)
		runningTasks []task.Info
	)

	for {
		pNo = pNo + 1
		if tInfo, err := manager.getTaskList(pNo, pSize, channel, true); err != nil {
			log.Err(fmt.Sprintf("获取运行队列(%v)数据失败:%+v", channel, err))
			continue
		} else {

			for _, v := range tInfo.List {

				item := v.(task.Info)
				runningTasks = append(runningTasks, item)
			}

			if tInfo.Pagination.PageSize < pSize {
				break
			}

		}
	}

	return runningTasks, nil
}

func (manager *TaskManagerService) readTask() {

	var (
		lastPrintTime  = time.Now().Unix()
		ctx            = context.Background()
		runningChannel = fmt.Sprintf("%v_running", manager.redisParams.Channel)
	)

	for {

		if manager.checkWaitingState() {
			time.Sleep(time.Millisecond * 5)
			continue
		}

		/* 尾取 */
		result, err := manager.redisCli.BRPopLPush(ctx, manager.redisParams.Channel, runningChannel, time.Second*1).Result()
		if err != nil {
			now := time.Now().Unix()
			if now > lastPrintTime+600 {
				lastPrintTime = now
				log.Info(fmt.Sprintf("从 REDIS LIST (%v) 读取筛料任务数据失败,可能已经过了超时时间:%+v", manager.redisParams, err))
			}
			continue
		}

		taskId := result
		result, err = manager.redisCli.HGet(ctx, manager.redisParams.DetailChannel, taskId).Result()
		if err != nil {
			log.Err(fmt.Sprintf("从 REDIS HASH(%v) 读取筛料任务数据失败:%+v", manager.redisParams, err))
			continue
		}

		var taskInfo task.Info
		if err := json.Unmarshal([]byte(result), &taskInfo); err != nil {
			log.Err(fmt.Sprintf("从 REDIS(%v) 解析筛料任务数据失败:%+v", manager.redisParams, err))
			continue
		}

		executor, ok := execute.Get(taskInfo.Operator)
		if !ok {
			log.Err(fmt.Sprintf("未注册或者没有实现的业务类型 %+v", taskInfo.Operator))
			/* 需要将错误的列表写入到错误列表 */
			continue
		}

		taskService := manager.createServiceExecutor(taskInfo)
		if err := manager.Start(taskId, true); err != nil {
			log.Err(fmt.Sprintf("任务:%v 设置任务状态为 启动 失败:%+v", taskId, err))
			continue
		}

		_ = manager.run(executor, taskService)
	}
}

func (manager *TaskManagerService) run(executor execute.TaskExecutorInterface, taskService *TaskExecutorService) error {

	/*
		对执行情况做局部监控，并做逻辑动态调整
	*/
	wp := sync.WaitGroup{}
	wp.Add(2)

	var (
		start         = time.Now()
		lastPrintTime = time.Now().Unix()
		taskId        = taskService.taskInfo.Id
		taskFinished  = atomic.Bool{}
	)

	go func() {
		defer wp.Done()

		for {

			//
			if taskService.checkFinishState(manager) || taskService.checkTerminateState(manager) {
				log.Info(fmt.Sprintf("执行 筛料任务:%v 结束,任务状态(应该为已完成或者被终止)为:%+v", taskService.taskInfo.Id, taskService.taskInfo))
				break
			}

			if taskFinished.Load() {
				log.Info(fmt.Sprintf("执行 筛料任务:%v 结束,任务状态(应该为自然完成)为:%+v", taskService.taskInfo.Id, taskService.taskInfo))
				break
			}

			if taskService.checkPauseState(manager) {

				now := time.Now().Unix()
				if now > lastPrintTime+60 {
					lastPrintTime = now
					log.Info(fmt.Sprintf("执行 筛料任务:%v 结束,任务状态(应该为已暂停)为:%+v", taskService.taskInfo.Id, taskService.taskInfo))
				}

				time.Sleep(time.Millisecond * 5)
				continue
			}

			if dispatchCount := taskService.GetDispatchCount(); dispatchCount != nil && *dispatchCount != 0 {

				now := time.Now().Unix()
				if now > lastPrintTime+60 {
					lastPrintTime = now
					log.Info(fmt.Sprintf("执行 筛料任务:%v 结束,任务状态(应该为已暂停之后的启动状态，或者启动状态)为:%+v 等待上次因暂停而没有落地的料子完成 %+v",
						taskService.taskInfo.Id, taskService.taskInfo, *dispatchCount))
				}

				time.Sleep(time.Millisecond * 5)
				continue
			}

			if err := taskService.run(executor, manager); err != nil {
				log.Err(fmt.Sprintf("任务:%v 执行筛料失败:%+v", taskId, err))
			}
			log.Info(fmt.Sprintf("执行 筛料任务:%v 结束,任务状态(状态未知)为:%+v", taskService.taskInfo.Id, taskService.taskInfo))
		}
	}()

	go func() {
		defer wp.Done()
		lpt := time.Now().Unix()
		for progress := range taskService.readProgress() {

			if progress.Results() == nil {
				newValue := taskService.AddDispatchCount(-int32(progress.NotifyCount))
				now := time.Now().Unix()
				if newValue == 0 || now >= lpt+30 {
					lpt = now
					log.Info(fmt.Sprintf("任务ID:%v dispatch new:%+v", taskService.taskInfo.Id, newValue))
				}
			} else {
				if err := manager.writeResult(taskService, progress); err != nil {
					log.Err(fmt.Sprintf("筛料任务:%v 将任务:%+v 落地到redis失败:%+v", taskService.taskInfo.Id, taskService.taskInfo, err))
				} else {
					newValue := taskService.AddDispatchCount(-int32(progress.NotifyCount))
					now := time.Now().Unix()
					if newValue == 0 || now >= lpt+30 {
						lpt = now
						log.Info(fmt.Sprintf("任务ID:%v dispatch new:%+v", taskService.taskInfo.Id, newValue))
					}
				}
			}
		}
		taskFinished.Store(true)
		log.Info(fmt.Sprintf("收到 筛料任务:%v 终止信号", taskService.taskInfo.Id))
	}()

	go func() {
		wp.Wait()
		var (
			duration = time.Since(start)
			calc     = taskService.counter.Calculate(int(duration.Seconds()))
		)

		log.Info(fmt.Sprintf("任务:%v 已经执行完成,耗时:%v 统计数据(%v)", taskId, duration.String(), calc.String()))

		if err := manager.finish(taskId); err != nil {
			log.Err(fmt.Sprintf("删除任务(%v)失败:%+v", taskId, err))
		}

	}()

	return nil
}

func (manager *TaskManagerService) writeResult(service *TaskExecutorService, taskResult *task.Result) error {

	var (
		chunkSize   = 200
		retryIndex  int
		retryCount  = 10
		ctx         = context.Background()
		tx          = manager.redisCli.TxPipeline()
		values      = taskResult.SplitResult(chunkSize)
		phoneValues = taskResult.SplitPhones(chunkSize)
	)

	resultKey := fmt.Sprintf("%v", service.taskInfo.ResultListKey)
	for _, value := range values {
	retry1:
		if err := tx.LPush(ctx, resultKey, value).Err(); err != nil && retryIndex < retryCount {
			sleepDur := time.Millisecond * time.Duration(10*retryIndex)
			time.Sleep(sleepDur)
			retryIndex = retryIndex + 1
			log.Err(fmt.Sprintf("任务ID:%v 提交 Redis 事务(提交任务结果)超时(%v), 休眠%v 继续重试，第%v次重试,总共重试次数:%+v",
				service.taskInfo.Id, err, sleepDur.String(), retryIndex, retryCount))
			goto retry1
		}
	}

	retryIndex = 0
retry2:
	srcKey := fmt.Sprintf("%v", service.taskInfo.SrcListKey)
	for _, value := range phoneValues {
		if err := tx.SRem(ctx, srcKey, value).Err(); err != nil && retryIndex < retryCount {
			sleepDur := time.Millisecond * time.Duration(10*retryIndex)
			time.Sleep(sleepDur)
			retryIndex = retryIndex + 1
			log.Err(fmt.Sprintf("任务ID:%v 提交 Redis 事务(删除key:%v)超时(%v), 休眠%v 继续重试，第%v次重试,总共重试次数:%+v",
				service.taskInfo.Id, srcKey, err, sleepDur.String(), retryIndex, retryCount))
			goto retry2
		}
	}

	retryIndex = 0
retry3:

	detailChannel := fmt.Sprintf("%v", manager.redisParams.DetailChannel)
	t := time.Now()
	tInfo := service.taskInfo
	tInfo.UpdateAt = t
	tInfo.ElapsedTime = t.Sub(tInfo.CreateAt).String()
	tInfo.Progress = service.progress.Progress
	mapValue := map[string]interface{}{
		tInfo.Id: tInfo.String(),
	}

	if err := tx.HMSet(context.Background(), detailChannel, mapValue).Err(); err != nil && retryIndex < retryCount {
		sleepDur := time.Millisecond * time.Duration(10*retryIndex)
		time.Sleep(sleepDur)
		retryIndex = retryIndex + 1
		log.Err(fmt.Sprintf("任务ID:%v 提交 Redis 事务(提交任务状态)超时, 休眠%v 继续重试，第%v次重试,总共重试次数:%+v", service.taskInfo.Id, sleepDur.String(), retryIndex, retryCount))
		goto retry3
	}

	retryIndex = 0

retry4:
	execResults, err := tx.Exec(context.Background())
	if err != nil {
		if err == redis.TxFailedErr && retryIndex < retryCount {
			sleepDur := time.Millisecond * time.Duration(10*retryIndex)
			time.Sleep(sleepDur)
			retryIndex = retryIndex + 1
			log.Err(fmt.Sprintf("任务ID:%v 提交 Redis 事务超时, 休眠%v 继续重试，第%v次重试,总共重试次数:%+v", service.taskInfo.Id, sleepDur.String(), retryIndex, retryCount))
			goto retry4
		}
		_ = tx.Discard()
		log.Err(fmt.Sprintf("提交Redis事务失败:%+v,", err))
		return err
	}

	for _, result := range execResults {
		if err := result.Err(); err != nil {
			_ = tx.Discard()
			log.Err(fmt.Sprintf("任务ID:%v,将任务结果写入到 redis 消息队列失败（%+v），执行事务回滚", service.taskInfo.Id, err))
			return err
		}
	}

	return nil
}

func (manager *TaskManagerService) del(taskId string) error {

	/* 此处需要删除 redis 任务队列(备份队列)中的任务 */
	manager.locker.Lock()
	defer manager.locker.Unlock()
	_, ok := manager.executors[taskId]
	if ok {
		delete(manager.executors, taskId)
	}

	go func() {
		if err := manager.delRedisTask(taskId); err != nil {
			log.Err(fmt.Sprintf("任务ID:%v 删除 redis 任务失败:%+v", taskId, err))
		}
	}()
	return nil
}

func (manager *TaskManagerService) dumpResultToFile(taskId string) error {

	manager.locker.Lock()
	executor, ok := manager.executors[taskId]
	if !ok {
		return errors.New("不存在的任务ID")
	}

	manager.locker.Unlock()

	return executor.dumpResultToFile()
}

// Pause /* 暂停: 是不再继续筛料了，那么就需要将任务停止，清理任务的相关数据，设置任务状态为已完成 */
func (manager *TaskManagerService) delRedisTask(taskId string) error {

	var (
		ctx            = context.Background()
		detailChannel  = fmt.Sprintf("%v", manager.redisParams.DetailChannel)
		runningChannel = fmt.Sprintf("%v_running", manager.redisParams.Channel)
	)

	_, err := manager.redisCli.Pipelined(ctx, func(pipeliner redis.Pipeliner) error {

		if err := manager.redisCli.HDel(ctx, detailChannel, taskId).Err(); err != nil {
			return err
		}

		if err := manager.redisCli.LRem(ctx, runningChannel, 1, taskId).Err(); err != nil {
			return err
		}
		return nil
	})

	return err
}

func (manager *TaskManagerService) setRedisState(tInfo *task.Info, taskId string, status task.Status) error {

	var (
		ctx           = context.Background()
		detailChannel = fmt.Sprintf("%v", manager.redisParams.DetailChannel)
	)

	/* tInfo 为空可能是还没有执行的任务（init状态的任务） */
	if tInfo == nil {
		if result, err := manager.redisCli.HGet(ctx, detailChannel, taskId).Result(); err != nil {
			return err
		} else {

			var (
				info = &task.Info{}
			)
			_ = json.Unmarshal([]byte(result), info)
			tInfo = info
		}
	}

	tInfo.Progress.State = status
	mapValue := map[string]interface{}{
		tInfo.Id: tInfo.String(),
	}

	if err := manager.redisCli.HMSet(ctx, detailChannel, mapValue).Err(); err != nil {
		return err
	}
	return nil
}

func (manager *TaskManagerService) getRedisState(taskId string) (*task.Info, error) {

	var (
		ctx           = context.Background()
		tInfo         = &task.Info{}
		detailChannel = fmt.Sprintf("%v", manager.redisParams.DetailChannel)
	)

	if result, err := manager.redisCli.HGet(ctx, detailChannel, taskId).Result(); err != nil {
		return nil, err
	} else {

		var (
			info = &task.Info{}
		)
		_ = json.Unmarshal([]byte(result), info)
		tInfo = info
	}
	return tInfo, nil
}

func (manager *TaskManagerService) getState(taskId string) (*task.Info, error) {

	var (
		tInfo = &task.Info{}
	)

	// manager.locker.Lock()
	// defer manager.locker.Unlock()
	//
	// _service, ok := manager.executors[taskId]
	// if ok {
	//
	//	tInfo = &_service.taskInfo
	//	tInfo.TaskProgress = _service.progress.TaskProgress
	//
	//	return tInfo, nil
	// }

	if t, err := manager.getRedisState(taskId); err != nil {

		log.Err(fmt.Sprintf("获取任务:%v 状态失败:%+v", taskId, err))

	} else {
		tInfo = t
	}

	return tInfo, nil
}

func (manager *TaskManagerService) setState(taskId string, status task.Status, f func(service *TaskExecutorService)) error {

	var (
		tInfo *task.Info
	)

	manager.locker.Lock()
	_service, ok := manager.executors[taskId]
	if ok {
		f(_service)

		tInfo = &_service.taskInfo
		tInfo.Progress = _service.progress.Progress
		manager.executors[taskId].taskInfo = *tInfo
	}

	manager.locker.Unlock()

	if err := manager.setRedisState(tInfo, taskId, status); err != nil {

		log.Err(fmt.Sprintf("设置任务:%v 状态失败", taskId))

	}

	return nil
}

func (manager *TaskManagerService) Reboot() {

}

func (manager *TaskManagerService) SetConcurrency(concurrency int) {

	defer func() {

		/* 1. 集群通知 */
		_ = manager.clusterServer.Set("concurrency", concurrency)

		/* 2. 更新 redis */
		_ = setTaskConfig(manager.taskConfig)
	}()

	manager.locker.Lock()
	defer manager.locker.Unlock()

	manager.taskConfig.concurrency = int32(concurrency)
}

func (manager *TaskManagerService) index(taskId string) error {

	// var (
	//	ctx = context.Background()
	// )
	//
	// if index, err := manager.redisCli.RPush(ctx, manager.redisParams.Channel, taskId, redis.LPosArgs{}).Result(); err != nil {
	//	return err
	// }
	// manager.redisCli.LMove()
	// manager.redisCli.LIndex(ctx, manager.redisParams.Channel)

	return nil
}

/*
1. 只可调整排队（manager.redisParams.Channel）中的任务的优先级
2. 运行的队列的优先级不可调整
*/
func (manager *TaskManagerService) AdjustPriority(fromId, toId string, beforeOrAfter int) error {

	manager.taskOp.Store(true)
	defer manager.taskOp.Store(false)

	var (
		ctx = context.Background()
	)
	_, err := manager.redisCli.Pipelined(ctx, func(pipeLiner redis.Pipeliner) error {

		fInsert := pipeLiner.LInsertBefore
		switch beforeOrAfter {
		case 1:
			fInsert = pipeLiner.LInsertBefore
			break
		case 2:
			fInsert = pipeLiner.LInsertAfter
		default:
			fInsert = pipeLiner.LInsertBefore
			break
		}

		if err := pipeLiner.LRem(ctx, manager.redisParams.Channel, 1, fromId).Err(); err != nil {
			return err
		}

		if err := fInsert(ctx, manager.redisParams.Channel, toId, fromId).Err(); err != nil {
			return err
		}

		return nil
	})

	return err
}

func (manager *TaskManagerService) GetTaskList(state int, pNo, pSize int, detail bool) (*TaskInfoWithPagination, error) {

	/*
		1. state = 1 获取运行中的任务列表(也可以从 redis 中获取)
		2. state = 2 获取排队中的任务列表(从 redis 中获取)
		3. 不传或者 state = 0 获取所有的任务列表
	*/

	channel := manager.redisParams.Channel
	if state == 1 {
		channel = fmt.Sprintf("%v_running", channel)
	}

	if state == 2 {
		manager.taskOp.Store(true)
		defer manager.taskOp.Store(false)
	}

	return manager.getTaskList(pNo, pSize, channel, detail)
}

func (manager *TaskManagerService) getTaskList(pNo, pSize int, channel string, detail bool) (*TaskInfoWithPagination, error) {

	var (
		pageNo   = 1
		pageSize = 500
	)
	if pNo != 0 {
		pageNo = pNo
	}

	if pSize != 0 {
		pageSize = pSize
	}

	var (
		taskInfos = &TaskInfoWithPagination{
			Pagination: &PaginationResult{
				PageIndex: pageNo,
			},
		}
		taskDetail = manager.redisParams.DetailChannel
		ctx        = context.Background()
		errs       []string
	)

	// 计算起始索引和结束索引
	startIdx, endIdx := int64((pageNo-1)*pageSize), int64(pageNo*pageSize-1)

	results, err := manager.redisCli.LRange(ctx, channel, startIdx, endIdx).Result()
	if err != nil {
		return nil, err
	}

	if len(results) > 0 {
		if detail {
			if details, err := manager.redisCli.HMGet(ctx, taskDetail, results...).Result(); err != nil {
				return nil, err
			} else {

				for _, item := range details {
					var info = task.Info{}
					_ = json.Unmarshal([]byte(fmt.Sprintf("%v", item)), &info)
					taskInfos.List = append(taskInfos.List, info)
				}
			}

		} else {

			for _, item := range results {
				var ele interface{} = item
				taskInfos.List = append(taskInfos.List, ele)
			}

		}
	}

	taskInfos.Pagination.Total = manager.redisCli.LLen(ctx, channel).Val()
	taskInfos.Pagination.PageSize = len(results)

	if len(errs) == 0 {
		return taskInfos, nil
	}
	return taskInfos, errors.New(strings.Join(errs, ","))
}

func (manager *TaskManagerService) GetClusterList(data GetClusterListReq) (interface{}, error) {

	return manager.clusterServer.cluster.Members(), nil
}
